package com.new4net.reactor;

import com.new4net.common.Tuple;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class InitiationDispatcher {
    //关注的主题
    private Subject subject;

    private volatile long maxSleep = 500;

    private volatile int maxResult = 5000;

    private volatile boolean runable = true;


    private volatile double minWorkRate = 0.75;

    private Map<Object,List<EventHandler>> eventHandlers = new ConcurrentHashMap<>();

    protected List<Command> executingCommand =  new ArrayList<>();

    protected volatile List<Command> executedCommand ;

    public long getMaxSleep() {
        return maxSleep;
    }

    public int getMaxResult() {
        return maxResult;
    }

    public boolean isRunable() {
        return runable;
    }

    public double getMinWorkRate() {
        return minWorkRate;
    }


    public List<Command> getExecutingCommand() {
        return executingCommand;
    }

    public List<Command> getExecutedCommand() {
        return executedCommand;
    }

    protected Demultiplexer demultiplexer = new Demultiplexer();

    public InitiationDispatcher(Subject subject) {
        this.subject = subject;
    }

    public void registerEventHander(String event, EventHandler eventHandler) {
        List<EventHandler> handlers;
        synchronized (eventHandlers){
            handlers   = eventHandlers.get(event);
            if(handlers==null){
                handlers = new Vector<>();
                eventHandlers.put(event,handlers);
            }


        }
        handlers.add(eventHandler);
    }

    public void removeEventHandler(String event,EventHandler eventHandler) {
        List<EventHandler> handlers;
        synchronized (eventHandlers){
            handlers  = eventHandlers.get(event);

        }

        if(handlers !=null){
            handlers.remove(eventHandler);
        }

    }

    public void stop() {
        runable = false;
    }

    public void handleEvents() {
        log.info("This dispatcher is running");
        runable = true;

        dispatch();

        log.info("This dispatcher is stopped");

    }

    private void dispatch() {
        while (runable) {
            sleepAnyTime(work());
        }
    }

    private Tuple work() {
        long startTime = System.nanoTime();
        List<Command> commands = demultiplexer.select(subject,eventHandlers);
        for (Command command : commands) {
            executingCommand.add(command);
            handleEvent(command);

        }
        handleResult();
        return new Tuple<>(System.nanoTime() - startTime, executingCommand.size());
    }

    protected void handleResult() {
        List<Command> oldExecutedCommand = new ArrayList<>();
        if(executedCommand!=null){
            //copy
            Iterator<Command> iterator = executedCommand.iterator();
            while (iterator.hasNext()) {
                oldExecutedCommand.add(iterator.next());
            }
        }



        //check new completed command
        Iterator<Command> iterator = executingCommand.iterator();
        while (iterator.hasNext()) {
            Command next = iterator.next();
            if (next.getEndTime() != null) {
                oldExecutedCommand.add(next);
                iterator.remove();
            }
        }


        int index = oldExecutedCommand.size() - maxResult;
        if (index > 0) {
            //discard longger command
            List<Command> newExecutedCommand = new ArrayList<>();
            for (int i = 0; i < maxResult; i++) {
                newExecutedCommand.add(oldExecutedCommand.get(i + index));

            }
            executedCommand = newExecutedCommand;
        } else {
            executedCommand = oldExecutedCommand;

        }


    }

    protected void sleepAnyTime(Tuple<Long, Integer> workParams) {

        long n = workParams.getT2();
        if (n <= 0) {
            try {
                Thread.sleep(new Random().nextInt(300) + 200);
            } catch (InterruptedException e) {
                log.info("I wake and work again");
            }
        } else {
            Long t1 = workParams.getT1();
            if (n <= Runtime.getRuntime().availableProcessors()) {
                try {
                    double v = t1.doubleValue();
                    long l = (long) (v / minWorkRate - v);
                    l = l >>> 10;
                    if (l > 0) {
                        Thread.sleep(l > maxSleep ? maxSleep : l);
                    }
                } catch (InterruptedException e) {
                    log.info("I wake and work again");
                }
            }
        }

    }



    public void setMaxSleep(long maxSleep) {
        this.maxSleep = maxSleep;
    }

    public void setMaxResult(int maxResult) {
        this.maxResult = maxResult;
    }

    /**
     * 最低工作效率
     *
     * @param minWorkRate
     */

    public void setMinWorkRate(double minWorkRate) {
        this.minWorkRate = minWorkRate;
    }

    protected void handleEvent(Command command) {
        command.run();
    }


}
